package com.leilei.broadcaststate;

import com.leilei.broadcaststate.common.Location;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/25 21:36
 */
public class LocationProcessFunction extends BroadcastProcessFunction<Location, Tuple2<Integer, List<Integer>>, Location> {
    private MapStateDescriptor<Integer, List<Integer>> descriptor;

    public LocationProcessFunction(MapStateDescriptor<Integer, List<Integer>> descriptor) {
        this.descriptor = descriptor;
    }

    @Override
    public void processElement(Location value, ReadOnlyContext ctx, Collector<Location> out) throws Exception {
        Integer vehicleId = value.getVehicleId();
        //根据状态描述器获取广播状态
        ReadOnlyBroadcastState<Integer, List<Integer>> broadcastState = ctx.getBroadcastState(descriptor);
        if (broadcastState != null) {
            //取出广播状态中的map<用户id,<姓名,年龄>>
            List<Integer> orgList = broadcastState.get(vehicleId);
            if (orgList != null && !orgList.isEmpty()) {
                value.setOrgIdList(orgList);
                out.collect(value);
            }
        }
    }

    @Override
    public void processBroadcastElement(Tuple2<Integer, List<Integer>> value, Context ctx, Collector<Location> out) throws Exception {
        //先根据状态描述器获取历史的广播状态
        BroadcastState<Integer, List<Integer>> broadcastState = ctx.getBroadcastState(descriptor);
        //最后将最新的广播流数据放到state中（更新状态数据）
        broadcastState.put(value.f0, value.f1);
    }


}
